package com.storm.queue

import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.locks.Condition
import java.util.concurrent.locks.ReentrantLock
import java.util.function.LongUnaryOperator
import kotlin.math.log

/**
 * 延迟队列全局参数
 */
object RedisQueueKeys {
    /**
     * 是一个Hash_Table结构；里面存储了所有的延迟队列的信息;KV结构；
     * K=TOPIC:ID    V=CONENT;  V由客户端传入的数据,消费的时候回传；
     */
    const val REDIS_DELAY_TABLE = "Redis_Delay_Table"

    /**
     * 延迟队列的有序集合; 存放K=TOPIC:ID 和需要的执行时间戳;
     * 根据时间戳排序;
     */
    const val RD_ZSET_BUCKET_PRE = "RD_ZSET_BUCKET:"

    /**
     * list结构; 每个Topic一个list；list存放的都是当前需要被消费的Job;
     */
    const val RD_LIST_TOPIC_PRE = "RD_LIST_TOPIC:"

    const val RD_LIST_TOPIC_PRE_LOCK = "bucket_z_set"

}


/**
 * 超时机制
 */
object TimeoutUtil {
    val log: Logger = LoggerFactory.getLogger(TimeoutUtil.javaClass)
    /**
     * 执行用户回调接口的 线程池;
     * 计算回调接口的超时时间
     */
    private val executorService = Executors.newCachedThreadPool()

    /**
     * 有超时时间的方法
     * @param timeout 时间秒
     */
    @Throws(InterruptedException::class, ExecutionException::class, TimeoutException::class)
    fun timeoutMethod(timeout: Long, callable: Callable<Boolean>) {
        val futureTask: FutureTask<Boolean> = FutureTask(callable)
        executorService.execute(futureTask)
        try {
            futureTask[timeout, TimeUnit.MILLISECONDS]
        } catch (e: InterruptedException) {
            log.error(e.message, e)
            futureTask.cancel(true)
            throw e
        } catch (e: ExecutionException) {
            log.error(e.message, e)
            futureTask.cancel(true)
            throw e
        } catch (e: TimeoutException) {
            log.error(e.message, e)
            futureTask.cancel(true)
            throw e
        }
    }
}

/**用于synchronized加锁 */
val lock = ReentrantLock()
val condition: Condition = lock.newCondition()

object NextTimeHolder{

    var nextTime = AtomicLong(0)
    /**
     * 此方法为线程安全
     * 尝试将nextTime修改为newTime;
     * 因为是循环CAS操作,修改成功为止
     * 所以如果newTime< nextTime则修改为newTime
     * 否则还是修改为nextTime;
     * 如果修改之后的值是newTime ;则需要通知一下LockUtil.lock 醒过来;
     * 如果仍旧是nextTime说明要么当前的执行时间很后面或者说被另外一个线程修改为了更小的值
     * @param newTime  新的执行时间
     */
    fun tryUpdate(newTime: Long) {
        val updateFunction = LongUnaryOperator { _: Long ->
            if (newTime < nextTime.get())
                newTime
            else
                nextTime.get()
        }
        val next = nextTime.updateAndGet(updateFunction)

        //如果最终更新的不是  newTime说明已经被其他线程更新为更小的数了;
        if (next == newTime) {
            lock.lock()
            try{
                condition.signalAll()
            }finally {
                lock.unlock()
            }
        }
    }
    /**
     * 将nextTime设置为0之后  并且通知  就会立马执行一次搬运操作
     */
    fun setZeroAndNotify() {
        lock.lock()
        try {
            nextTime.set(0)
            condition.signalAll()
        }finally {
            lock.unlock()
        }
    }
}


class RedisKeyUtil(private val objectName: String){
    /**
     * 获取RD_LIST_TOPIC 的Key前缀
     * @return
     */
    fun getTopicListPreKey(): String {
        return objectName + ":" + RedisQueueKeys.RD_LIST_TOPIC_PRE
    }

    /**
     * 获取RD_LIST_TOPIC某个TOPIC的 Key
     * @param topic
     * @return
     */
    fun getTopicListKey(topic: String): String {
        return getTopicListPreKey() + topic
    }

    /**
     * 从member中解析出TopicList的key
     * @param member
     * @return
     */
    fun getTopicListKeyByMember(member: String): String {
        return getTopicListKey(getTopicKeyByMember(member))
    }

    /**
     * 拼接TOPIC:ID
     * @param topic
     * @return
     */
    fun getTopicId(topic: String, id: String): String {
        return "$topic:$id"
    }


    /**
     * 获取所有Job数据存放的Hash_Table 的Key
     * @return
     */
    fun getDelayQueueTableKey(): String {
        return objectName + ":" + RedisQueueKeys.REDIS_DELAY_TABLE
    }

    /**
     * 获取延迟列表 ZSet的Key
     * @return
     */
    fun getBucketKey(): String {
        return objectName + ":" + RedisQueueKeys.RD_ZSET_BUCKET_PRE
    }


    /**
     * 根据member获取Topic
     * @param member
     * @return
     */
    fun getTopicKeyByMember(member: String): String {
        val s = member.split(":".toRegex()).toTypedArray()
        return s[0]
    }
}



